package edu.hcmus.sow.dao;

import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map;

import me.prettyprint.cassandra.serializers.SerializerTypeInferer;
import me.prettyprint.cassandra.service.template.ColumnFamilyResult;
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;
import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.Serializer;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.ddl.ColumnDefinition;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

import org.springframework.stereotype.Repository;

@Repository
public class AbstractCassandraDAO<K extends Object, N extends Object, V extends Object> {
   protected Serializer<K> keySerializer;
   protected Serializer<N> nameSerializer;
   protected Serializer<V> valueSerializer;

   private static final String CLUSTER_NAME = "Test Cluster";
   private static final String CASSANDRA_HOST = "localhost:9160";

   // TODO: Load from configuration

   @SuppressWarnings("unchecked")
   public AbstractCassandraDAO() {
      if (getClass().getGenericSuperclass() instanceof ParameterizedType) {
         this.keySerializer = (Serializer<K>) getSerializer(0);
         this.nameSerializer = (Serializer<N>) getSerializer(1);
         this.valueSerializer = (Serializer<V>) getSerializer(2);
      }
   }

   private Serializer<?> getSerializer(int index) {
      Class<?> klass = (Class<?>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[index];
      return SerializerTypeInferer.getSerializer(klass);
   }

   /**
    * Persist a column to Cassandra.
    * 
    * @param keyspace
    * @param columnFamily
    * @param key
    * @param name
    * @param value
    */
   public void persist(String keyspace, String columnFamily, K key, N name, V value) {
      HColumn<N, V> column = HFactory.createColumn(name, value, nameSerializer, valueSerializer);
      createMutator(keyspace, columnFamily).addInsertion(key, columnFamily, column).execute();
   }

   /**
    * Persist a row to Cassandra.
    * 
    * @param keyspace
    * @param columnFamily
    * @param key
    * @param row
    */
   public void persist(String keyspace, String columnFamily, K key, Map<N, V> row) {
      Mutator<K> mutator = createMutator(keyspace, columnFamily);
      for (N columnName : row.keySet()) {
         HColumn<N, V> column = HFactory.createColumn(columnName, row.get(columnName), nameSerializer, valueSerializer);
         mutator.addInsertion(key, columnFamily, column);
      }
      mutator.execute();
   }

   /**
    * Delete a row from Cassandra.
    * 
    * @param keyspace
    * @param columnFamily
    * @param key
    */
   public void delete(String keyspace, String columnFamily, K key) {
      createMutator(keyspace, columnFamily).addDeletion(key, columnFamily).execute();
   }

   /**
    * Fetch a column from Cassandra.
    * 
    * @param keyspace
    * @param columnFamily
    * @param key
    * @param name
    * @return value
    */
   public V fetchColumn(String keyspace, String columnFamily, K key, N name) {
      HColumn<N, V> column = createTemplate(keyspace, columnFamily).querySingleColumn(key, name, valueSerializer);
      return column == null ? null : column.getValue();
   }

   /**
    * Fetch a row from Cassandra.
    * 
    * @param keyspace
    * @param columnFamily
    * @param key
    * @return Column name to column value Map
    */
   public Map<N, V> fetchRow(String keyspace, String columnFamily, K key) {
      Map<N, V> row = new LinkedHashMap<N, V>();
      ColumnFamilyResult<K, N> result = createTemplate(keyspace, columnFamily).queryColumns(key);
      for (N columnName : result.getColumnNames()) {
         V columnValue = valueSerializer.fromByteBuffer(result.getColumn(columnName).getValue());
         row.put(columnName, columnValue);
      }
      return row;
   }

   /**
    * Create column family.
    * 
    * @param keyspace
    * @param columnFamily
    */
   public void createColumnFamily(String keyspace, String columnFamily) {
      ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition(keyspace, columnFamily,
            keySerializer.getComparatorType(), new ArrayList<ColumnDefinition>());
      cfDef.setDefaultValidationClass(valueSerializer.getComparatorType().getClassName());
      cfDef.setKeyValidationClass(keySerializer.getComparatorType().getClassName());
      getCluster().addColumnFamily(cfDef);
   }

   /**
    * Drop column family.
    * 
    * @param keyspace
    * @param columnFamily
    */
   public void dropColumnFamily(String keyspace, String columnFamily) {
      getCluster().dropColumnFamily(keyspace, columnFamily);
   }

   protected Mutator<K> createMutator(String keyspace, String columnFamily) {
      Keyspace hectorKeyspace = HFactory.createKeyspace(keyspace, getCluster());
      return HFactory.createMutator(hectorKeyspace, keySerializer);
   }

   protected ColumnFamilyTemplate<K, N> createTemplate(String keyspace, String columnFamily) {
      Keyspace hectorKeyspace = HFactory.createKeyspace(keyspace, getCluster());
      return new ThriftColumnFamilyTemplate<K, N>(hectorKeyspace, columnFamily, keySerializer, nameSerializer);
   }

   protected Cluster getCluster() {
      return HFactory.getOrCreateCluster(CLUSTER_NAME, CASSANDRA_HOST);
   }
}